package waterMark;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;

public class waterMark {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        //获取数据源
        DataStreamSource<String> data = env.socketTextStream("linux121", 7777);
        SingleOutputStreamOperator<Tuple2<String, Long>> maped = data.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                String[] split = value.split(",");
                return new Tuple2<>(split[0], Long.valueOf(split[1]));
            }
        });
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = maped.assignTimestampsAndWatermarks(new WatermarkStrategy<Tuple2<String, Long>>() {
            @Override
            public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Tuple2<String, Long>>() {
                    private long maxTimeStamp = 0L;

                    @Override
                    public void onEvent(Tuple2<String, Long> evnt, long l, WatermarkOutput watermarkOutput) {
                        maxTimeStamp = Math.max(maxTimeStamp, evnt.f1);
                        System.out.println("maxTimeStamp:"+maxTimeStamp+"...format:"+sdf.format(maxTimeStamp));
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                       // System.out.println("....onPeriodicEmit...");
                        long maxOutOfOrderness = 3000l;

                        watermarkOutput.emitWatermark(new Watermark(maxTimeStamp - maxOutOfOrderness));
                    }
                };
            }
        }.withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
            @Override
            public long extractTimestamp(Tuple2<String, Long> element, long l) {
                return element.f1;
            }
        }));

        KeyedStream<Tuple2<String, Long>, String> keybyed = watermarks.keyBy(value -> value.f0);
        WindowedStream<Tuple2<String, Long>, String, TimeWindow> windowed = keybyed.window(TumblingEventTimeWindows.of(Time.seconds(4)));

        SingleOutputStreamOperator<String> result = windowed.apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
                //System.out.println("====================================");
                String key = s;
                ArrayList<Long> list = new ArrayList<>();
                Iterator<Tuple2<String, Long>> iterator = iterable.iterator();
                while (iterator.hasNext()) {
                    Tuple2<String, Long> next = iterator.next();
                    System.out.println(next.f1);
                    list.add(next.f1);
                }
                Collections.sort(list);

                String results = "key:" + key + "..." + "list的大小:" + list.size() + "...list的第一个元素:" + sdf.format(list.get(0)) + "...list的最后一个元素:" + list.get(list.size() - 1) +
                        "...window.start:" + sdf.format(window.getStart()) + "...window.end:" + sdf.format(window.getEnd());
                collector.collect(results);
            }
        });

        result.print();
        env.execute();
    }
}
